package com.apps.sdses.flink141.jar;

import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.util.LinkedList;

/**
 * author: codejiwei
 * date: 2023/8/12
 * desc: flink tumble window get last value
 **/
public class FlinkTumbleWindowLastValue {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String sql = "CREATE TABLE source_table (\n" +
                "    dim STRING,\n" +
                "    user_id BIGINT,\n" +
                "    price BIGINT,\n" +
                "    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),\n" +
                "    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen',\n" +
                "  'rows-per-second' = '1',\n" +
//                "  'number-of-rows' = '10',\n" +
                "  'fields.dim.length' = '5',\n" +
                "  'fields.user_id.min' = '1',\n" +
                "  'fields.user_id.max' = '10',\n" +
                "  'fields.price.min' = '1',\n" +
                "  'fields.price.max' = '100'\n" +
                ")";

        tableEnv.executeSql(sql);
        Table table = tableEnv.from("source_table");
        DataStream<Row> stream = tableEnv.toDataStream(table);

        stream
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))

//                .keyBy(x -> x.<Long>getFieldAs("user_id"))
//                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
//                .process(new ProcessWindowFunction<Row, Row, Long, TimeWindow>() {
//                    Long maxPrice;
//                    Row row;
//
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        this.maxPrice = 0L;
//                        this.row = new Row(1);
//                    }
//
//                    @Override
//                    public void process(Long s, Context context, Iterable<Row> elements, Collector<Row> out) throws Exception {
//
//                        for (Row element : elements) {
////                            System.out.println("======element:" + element.toString() + "======");
//                            Long price = element.<Long>getFieldAs("price");
//                            if (price > this.maxPrice) {
//                                this.maxPrice = price;
//                                this.row = element;
//                            }
//                        }
//
//                        out.collect(this.row);
//                    }
//                })
//                .print();
//        .reduce((r1, r2) -> r1)
//        .reduce((r1, r2) -> r2)
//        .reduce((r1, r2) -> r1.<Long>getFieldAs("price") > r2.<Long>getFieldAs("price") ? r1 : r2)

                .process(new ProcessAllWindowFunction<Row, Row, TimeWindow>() {
                    private ListStateDescriptor<Row> state = new ListStateDescriptor<>("listState", Row.class);
                    private LinkedList<Row> list;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        list = new LinkedList<>();
                        System.out.println("============= init LinkedList ================");
                    }
//
//                    @Override
//                    public void close() throws Exception {
//                        list.clear();
//                    }


                    @Override
                    public void clear(ProcessAllWindowFunction<Row, Row, TimeWindow>.Context context) throws Exception {
                        System.out.println("============= clear state ==============");
                        list.clear();
                    }

                    @Override
                    public void process(ProcessAllWindowFunction<Row, Row, TimeWindow>.Context context, Iterable<Row> elements, Collector<Row> out) throws Exception {
                        for (Row element : elements) {
                            System.out.println("==========" + element);
                            list.add(element);
                        }
                        System.out.println("------list size:" + list.size());
                        if (list.size() > 5) {
                            out.collect(list.getLast());
                        }
                    }
                })
                .print();


        env.execute();


    }

}
